maskfile.md (1)

102-166: Hardcoded service list and retry parameters remain.

The retry logic has been improved with better status messages (lines 136-137, 157), but several issues from previous reviews remain:

1. Service list hardcoded (line 118): Adding/removing services requires code changes
2. Retry parameters not externally configurable: Variables exist (lines 122-124) but values are hardcoded
3. Status check incomplete: Only verifies ACTIVE status; doesn't confirm runningCount >= desiredCount

Consider these improvements:

🔎 Proposed refactor

+# Allow override via environment
+SERVICES="${PSF_SERVICES:-pocketsizefund-datamanager pocketsizefund-portfoliomanager pocketsizefund-equitypricemodel}"
+MAX_RETRIES="${PSF_MAX_RETRIES:-12}"
+RETRY_WAIT_SECONDS="${PSF_RETRY_WAIT_SECONDS:-5}"
+
-    for SERVICE in pocketsizefund-datamanager pocketsizefund-portfoliomanager pocketsizefund-equitypricemodel; do
+    for SERVICE in $SERVICES; do
         echo "Checking if $SERVICE exists and is ready"
 
-        # Wait up to 60 seconds for service to be active
-        RETRY_COUNT=0
-        MAX_RETRIES=12
-        RETRY_WAIT_SECONDS=5
-        SERVICE_READY=false
+        RETRY_COUNT=0
+        SERVICE_READY=false
 
         while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do
-            SERVICE_STATUS=$(aws ecs describe-services \
+            SERVICE_INFO=$(aws ecs describe-services \
                 --cluster "$CLUSTER" \
                 --services "$SERVICE" \
-                --query 'services[0].status' \
-                --output text 2>/dev/null || echo "NONE")
+                --query 'services[0].[status,runningCount,desiredCount]' \
+                --output text 2>/dev/null || echo "NONE 0 0")
+            
+            read -r STATUS RUNNING DESIRED <<< "$SERVICE_INFO"
 
-            if [ "$SERVICE_STATUS" = "ACTIVE" ]; then
+            if [ "$STATUS" = "ACTIVE" ] && [ "$RUNNING" -ge "$DESIRED" ] && [ "$DESIRED" -gt 0 ]; then
                 SERVICE_READY=true
-                echo "Service $SERVICE is ACTIVE"
+                echo "Service $SERVICE is ACTIVE with $RUNNING/$DESIRED tasks running"
                 break
-            elif [ "$SERVICE_STATUS" = "NONE" ]; then
+            elif [ "$STATUS" = "NONE" ]; then
                 echo "Service not found, waiting ($((RETRY_COUNT + 1))/$MAX_RETRIES)"
             else
-                echo "Service status: $SERVICE_STATUS, waiting ($((RETRY_COUNT + 1))/$MAX_RETRIES)"
+                echo "Service status: $STATUS ($RUNNING/$DESIRED tasks), waiting ($((RETRY_COUNT + 1))/$MAX_RETRIES)"
             fi

.github/workflows/launch_infrastructure.yaml (1)

27-50: Add error handling to prevent deployment with failed builds.

The separate build and push steps for each service run independently without error propagation. If an early build or push fails, the workflow continues and may deploy infrastructure with stale or missing images. Each flox/activate-action step should be configured to fail the workflow on error.

Verify that the flox/activate-action properly propagates exit codes, or add explicit verification steps between image operations and deployment:

      - name: Verify all images pushed
        run: |
          aws_account_id=$(aws sts get-caller-identity --query Account --output text)
          for image in datamanager portfoliomanager equitypricemodel; do
            if ! docker manifest inspect ${aws_account_id}.dkr.ecr.us-east-1.amazonaws.com/pocketsizefund/${image}-server:latest > /dev/null 2>&1; then
              echo "Error: Image ${image}-server:latest not found in ECR"
              exit 1
            fi
          done

tools/run_training_job.py (3)

13-13: Remove unused noqa directive.

The noqa: PLR0913 directive is unused according to Ruff, indicating that the "too many arguments" rule is not being triggered for this function. Removing unused directives keeps the code cleaner.

🔎 Proposed fix

-def run_training_job(  # noqa: PLR0913
+def run_training_job(
     application_name: str,
     aws_profile: str,
     image_uri: str,

---

34-41: Add structured logging before Estimator creation.

Adding a log statement before creating the Estimator would improve observability and help with debugging if the Estimator construction fails (e.g., invalid IAM role, malformed image URI).

🔎 Proposed enhancement

+    logger.info(
+        "Creating estimator",
+        image_uri=image_uri,
+        instance_type="ml.t3.xlarge",
+        output_path=output_path,
+    )
+
     estimator = Estimator(
         image_uri=image_uri,
         role=sagemaker_role,

---

49-57: Enhance logging for training execution.

The training job execution would benefit from more detailed structured logging to improve observability. Consider adding a log before starting the job and including additional context (S3 path, image URI, instance type) in the error log.

🔎 Proposed enhancement

     try:
+        logger.info(
+            "Starting training job execution",
+            s3_data_path=s3_data_path,
+            image_uri=image_uri,
+            instance_type="ml.t3.xlarge",
+        )
         estimator.fit({"train": training_data_input})
+        logger.info("Training job completed successfully")
     except Exception as e:
         logger.exception(
             "Error during training job",
             error=f"{e}",
             application_name=application_name,
+            s3_data_path=s3_data_path,
+            image_uri=image_uri,
         )
         raise RuntimeError from e

applications/equitypricemodel/src/equitypricemodel/tide_data.py (3)

29-30: Consider explicitly mentioning TiDE in the docstring.

The docstring describes "Time-series dense encoder" but doesn't explicitly state "TiDE" or expand the acronym. For better discoverability and clarity, consider: """TiDE (Time-series Dense Encoder) data preprocessing and postprocessing."""

🔎 Proposed enhancement

-    """Time-series dense encoder data preprocessing and postprocessing."""
+    """TiDE (Time-series Dense Encoder) data preprocessing and postprocessing."""

---

271-271: Remove unused noqa directives.

Lines 271, 273, and 276 have noqa directives for rules that aren't enabled in the Ruff configuration (E501, C901). These should be removed to keep the code clean.

🔎 Proposed fix

-            "decoder_continuous_features": 0,  # not using decoder_continuous_features for now  # noqa: E501
+            "decoder_continuous_features": 0,  # not using decoder_continuous_features for now

-            "static_continuous_features": 0,  # not using static_continuous_features for now  # noqa: E501
+            "static_continuous_features": 0,  # not using static_continuous_features for now

-    def get_batches(  # noqa: C901
+    def get_batches(

Also applies to: 273-273, 276-276

---

381-427: Use pathlib instead of os.path to align with coding guidelines and remove unused noqa directives.

The save/load methods correctly create the target directory (the previous os.path.dirname issue is fixed ✅). However, the code uses os.path with multiple unused noqa directives (PTH103, PTH118, PTH123). Switching to pathlib.Path aligns with Python best practices and the project's coding guidelines, eliminating the need for these suppressions.

🔎 Proposed refactor using pathlib

Add import at the top of the file:

+from pathlib import Path

Then refactor the save method:

     def save(self, directory_path: str) -> None:
-        os.makedirs(directory_path, exist_ok=True)  # noqa: PTH103
+        Path(directory_path).mkdir(parents=True, exist_ok=True)

-        with open(os.path.join(directory_path, "tide_data_mappings.json"), "w") as f:  # noqa: PTH118, PTH123
+        with (Path(directory_path) / "tide_data_mappings.json").open("w") as f:
             json.dump(self.mappings, f)

-        with open(os.path.join(directory_path, "tide_data_scaler.json"), "w") as f:  # noqa: PTH118, PTH123
+        with (Path(directory_path) / "tide_data_scaler.json").open("w") as f:

And the load method:

     @classmethod
     def load(cls, directory_path: str) -> "Data":
         data = cls()
-        with open(os.path.join(directory_path, "tide_data_mappings.json")) as f:  # noqa: PTH118, PTH123
+        with (Path(directory_path) / "tide_data_mappings.json").open() as f:
             data.mappings = json.load(f)

-        with open(os.path.join(directory_path, "tide_data_scaler.json")) as f:  # noqa: PTH118, PTH123
+        with (Path(directory_path) / "tide_data_scaler.json").open() as f:
             scaler_data = json.load(f)

Based on coding guidelines and past review feedback.

infrastructure/__main__.py (1)

116-128: Single NAT gateway creates availability zone dependency.

Using one NAT gateway in public_subnet_1 means all private subnet traffic egresses through availability zone A. If that AZ experiences issues, private tasks in AZ B lose internet connectivity.

For production, consider adding a second NAT gateway in public_subnet_2 with a separate route table for private_subnet_2. For now, this is acceptable for cost optimization.

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to **/*.py : Lint Python code with Ruff using the comprehensive ruleset

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to **/*.py : Use Ruff as the project formatter (replaces black/isort) for all Python code

applications/portfoliomanager/src/portfoliomanager/server.py (2)

258-262: Verify predictions endpoint uses parameterized queries.

This function forwards tickers_and_timestamps derived from prior_portfolio to the datamanager /predictions endpoint. Retrieved learnings indicate a known SQL injection vulnerability in applications/datamanager/src/predictions.rs where user-supplied tickers flow to storage::query_predictions_dataframe_from_s3 without validation, and SQL queries are built via string concatenation.

Confirm that the current datamanager /predictions handler uses parameterized queries or proper escaping for both tickers and timestamps to prevent exploitation.

Based on learnings, this vulnerability was acknowledged but deferred in PR #635.

#!/bin/bash
# Description: Check if datamanager predictions endpoint uses parameterized queries
# Expected: Find SQL query construction patterns in predictions handling

# Search for SQL query construction in datamanager predictions module
rg -nP --type=rust -C5 'query_predictions|SELECT.*FROM.*WHERE' applications/datamanager/

---

239-239: Fix typing.cast syntax error.

The first argument to cast must be a type, not a string. Line 239 passes "float" (a string literal), which will confuse Pyright and other type checkers.

🔎 Proposed fix

-    start_timestamp = datetime.fromtimestamp(cast("float", timestamps.min()), tz=UTC)
+    start_timestamp = datetime.fromtimestamp(cast(float, timestamps.min()), tz=UTC)

As per coding guidelines, type check with Pyright.

maskfile.md (1)

458-461: Environment loading pattern is sound; .env availability tracked separately.

The isolated subshell with set -a/set +a for environment variable export is a good practice. As previously discussed, the .env file will be provisioned via Flox hook.on-activate.

applications/datamanager/src/storage.rs (1)

186-194: Ticker validation addresses SQL injection; error message exposure previously acknowledged.

The ticker format validation effectively prevents SQL injection by allowing only alphanumeric characters, dots, and hyphens. The previous concern about the error message exposing the invalid ticker was acknowledged and deferred.

applications/equitypricemodel/src/equitypricemodel/tide_model.py (4)

1-15: LGTM—Structlog migration complete.

The file now correctly uses Structlog for structured logging, resolving the previous concern about Loguru usage.

---

197-253: LGTM—Training method improvements applied.

The train method now correctly:

• Preserves and restores Tensor.training state using try/finally (lines 204-251)
• Handles empty batch lists defensively (lines 235-239)
• Uses Structlog for structured logging

These changes resolve previous review concerns.

---

255-274: LGTM—Validation method handles edge cases.

The validate method now defensively handles empty validation batch lists by returning NaN with a warning (lines 270-272), preventing ZeroDivisionError.

---

276-316: LGTM—Persistence methods corrected.

The save method now correctly creates the target directory (line 280) instead of its parent, resolving the previous directory creation bug.

applications/equitypricemodel/src/equitypricemodel/tide_data.py (2)

29-30: LGTM—Docstring updated to reflect TiDE architecture.

The class docstring now correctly references "Time-series dense encoder" instead of the previous "Temporal fusion transformer", aligning with the PR's model migration.

---

381-427: LGTM—Persistence methods correctly implemented; remove unused noqa directives.

The save method now correctly creates the target directory (line 382) instead of its parent, resolving the previous bug. The load method properly reconstructs the Data instance with scaler and mappings.

However, Ruff flags several noqa pragmas as unused (lines 382, 384, 387, 411, 414). Consider removing these for cleaner code.

🔎 Optional cleanup

-        os.makedirs(directory_path, exist_ok=True)  # noqa: PTH103
+        os.makedirs(directory_path, exist_ok=True)

-        with open(os.path.join(directory_path, "tide_data_mappings.json"), "w") as f:  # noqa: PTH118, PTH123
+        with open(os.path.join(directory_path, "tide_data_mappings.json"), "w") as f:

-        with open(os.path.join(directory_path, "tide_data_scaler.json"), "w") as f:  # noqa: PTH118, PTH123
+        with open(os.path.join(directory_path, "tide_data_scaler.json"), "w") as f:
@@
-        with open(os.path.join(directory_path, "tide_data_mappings.json")) as f:  # noqa: PTH118, PTH123
+        with open(os.path.join(directory_path, "tide_data_mappings.json")) as f:

-        with open(os.path.join(directory_path, "tide_data_scaler.json")) as f:  # noqa: PTH118, PTH123
+        with open(os.path.join(directory_path, "tide_data_scaler.json")) as f:

.gitignore (1)

9-9: Minor: Redundant coverage ignore patterns.

Both coverage/ (line 9) and .coverage* (line 13) are now present in the file, which creates a small redundancy. The directory pattern coverage/ and file pattern .coverage* overlap in functionality. Consider whether both patterns are necessary or if one can be consolidated.

Also applies to: 13-13

tools/run_training_job.py (1)

13-20: Remove unused noqa directive.

Static analysis indicates PLR0913 is not enabled in the Ruff ruleset. The directive is unnecessary.

🔎 Proposed fix

-def run_training_job(  # noqa: PLR0913
+def run_training_job(
     application_name: str,

tools/sync_equity_bars_data.py (1)

102-102: Remove unused noqa directive.

Static analysis indicates PLR2004 is not enabled in the Ruff ruleset. The directive is unnecessary.

🔎 Proposed fix

-            if status_code >= 400:  # noqa: PLR2004
+            if status_code >= 400:

libraries/python/tests/test_equity_bars_schema.py (2)

192-208: Consider testing missing transactions column specifically.

While this test validates missing column behavior for open_price, it doesn't verify that the new transactions field is properly required. Since transactions is a newly added field (per the schema evolution), consider adding a dedicated test case to ensure missing transactions triggers the expected validation error.

🔎 Suggested additional test

Add a test case specifically for missing transactions:

def test_equity_bars_schema_missing_transactions() -> None:
    data = pl.DataFrame(
        {
            "ticker": ["AAPL"],
            "timestamp": [1674000000.0],
            "open_price": [150.0],
            "high_price": [155.0],
            "low_price": [149.0],
            "close_price": [153.0],
            "volume": [1000000],
            "volume_weighted_average_price": [152.5],
            # Missing transactions
        }
    )

    with pytest.raises((SchemaError, pl.exceptions.ColumnNotFoundError)):
        equity_bars_schema.validate(data)

---

7-247: Add validation tests for the transactions field.

The new transactions field is consistently added across all test cases, but there are no dedicated tests validating its constraints (e.g., negative values, zero values, null values). Following the established test patterns for other fields like volume and prices, consider adding comprehensive validation tests for transactions.

🔎 Suggested additional tests

def test_equity_bars_schema_negative_transactions() -> None:
    data = pl.DataFrame(
        {
            "ticker": ["AAPL"],
            "timestamp": [1674000000.0],
            "open_price": [150.0],
            "high_price": [155.0],
            "low_price": [149.0],
            "close_price": [153.0],
            "volume": [1000000],
            "volume_weighted_average_price": [152.5],
            "transactions": [-100],
        }
    )

    with pytest.raises(SchemaError):
        equity_bars_schema.validate(data)


def test_equity_bars_schema_zero_transactions() -> None:
    """Test that zero transactions is allowed (valid case with no trades)."""
    data = pl.DataFrame(
        {
            "ticker": ["AAPL"],
            "timestamp": [1674000000.0],
            "open_price": [150.0],
            "high_price": [155.0],
            "low_price": [149.0],
            "close_price": [153.0],
            "volume": [1000000],
            "volume_weighted_average_price": [152.5],
            "transactions": [0],
        }
    )

    # Adjust expectation based on schema requirements
    validated_df = equity_bars_schema.validate(data)
    assert validated_df["transactions"][0] == 0


def test_equity_bars_schema_null_transactions() -> None:
    data = pl.DataFrame(
        {
            "ticker": ["AAPL"],
            "timestamp": [1674000000.0],
            "open_price": [150.0],
            "high_price": [155.0],
            "low_price": [149.0],
            "close_price": [153.0],
            "volume": [1000000],
            "volume_weighted_average_price": [152.5],
            "transactions": [None],
        }
    )

    with pytest.raises(SchemaError):
        equity_bars_schema.validate(data)

Note: Adjust the zero transactions test expectation based on whether the schema should allow zero values (which might be valid for periods with no trading activity).

applications/portfoliomanager/src/portfoliomanager/server.py (1)

63-63: Remove unused noqa directive.

Ruff correctly flags that the noqa directive on line 63 is unnecessary since the codes PLR0911 and C901 are not enabled in the project's Ruff configuration.

🔎 Proposed fix

-def create_portfolio() -> Response:  # noqa: PLR0911, C901
+def create_portfolio() -> Response:

As per coding guidelines, Ruff is the project linter.

applications/datamanager/src/data.rs (1)

104-109: Consider combining chained with_columns calls for efficiency.

The sequential with_columns calls work correctly but could be combined into a single call for slightly better performance:

🔎 Proposed optimization

     let portfolio_dataframe = portfolio_dataframe
         .lazy()
-        .with_columns([col("ticker").str().to_uppercase().alias("ticker")])
-        .with_columns([col("side").str().to_uppercase().alias("side")])
-        .with_columns([col("action").str().to_uppercase().alias("action")])
+        .with_columns([
+            col("ticker").str().to_uppercase().alias("ticker"),
+            col("side").str().to_uppercase().alias("side"),
+            col("action").str().to_uppercase().alias("action"),
+        ])
         .collect()?;

infrastructure/__main__.py (1)

357-376: Misleading comment on portfoliomanager rule.

The comment on Line 361 states "Ensures that the more specific data manager paths take precedence" but is attached to the portfoliomanager_rule. Since ALB evaluates rules in ascending priority order and datamanager has priority 100 (evaluated first), the comment describes datamanager's behavior, not portfoliomanager's.

Move this comment to the datamanager_rule block (lines 378-401) or reword it to clarify the relationship.

applications/equitypricemodel/src/equitypricemodel/server.py (1)

31-125: Add structured logging to the prediction endpoint.

The endpoint performs multiple data retrieval, transformation, and model inference steps without logging. Adding structured log statements at key stages (data fetch, preprocessing, inference, save) would improve observability and debugging.

As per coding guidelines, use Structlog for structured logging across services.

Example structured logging additions

import structlog

logger = structlog.get_logger(__name__)

@application.post("/predictions")
def create_predictions() -> PredictionResponse:
    logger.info("Starting prediction generation")
    
    tide_model = Model.load(directory_path=".")
    logger.info("Model loaded successfully")
    
    # ... fetch data ...
    logger.info("Fetched equity data", bars_count=len(equity_bars_data), details_count=len(equity_details_data))
    
    # ... preprocessing ...
    logger.info("Data preprocessing complete", consolidated_rows=len(data))
    
    # ... inference ...
    logger.info("Generated predictions", batch_count=len(batches))
    
    # ... save ...
    logger.info("Predictions saved successfully")
    
    return PredictionResponse(data=processed_predictions.to_dict())

applications/equitypricemodel/src/equitypricemodel/tide_model.py (1)

95-95: Remove unused noqa directives flagged by Ruff.

Ruff reports several noqa pragmas as unused because the corresponding rules (PLR0913, PTH103, PTH118, PTH123) are not enabled in the current configuration. Removing these pragmas will clean up the code without changing behavior.

🔎 Lines to clean

-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
         input_size: int,
         hidden_size: int = 128,
@@
     def save(
         self,
         directory_path: str,
     ) -> None:
-        os.makedirs(directory_path, exist_ok=True)  # noqa: PTH103
+        os.makedirs(directory_path, exist_ok=True)

         states = get_state_dict(self)

-        safe_save(states, os.path.join(directory_path, "tide_states.safetensor"))  # noqa: PTH118
+        safe_save(states, os.path.join(directory_path, "tide_states.safetensor"))

         parameters = {
@@
-        parameters_file_path = os.path.join(directory_path, "tide_parameters.json")  # noqa: PTH118
-        with open(parameters_file_path, "w") as parameters_file:  # noqa: PTH123
+        parameters_file_path = os.path.join(directory_path, "tide_parameters.json")
+        with open(parameters_file_path, "w") as parameters_file:
             json.dump(parameters, parameters_file)

     @classmethod
     def load(
         cls,
         directory_path: str,
     ) -> "Model":
-        states_file_path = os.path.join(directory_path, "tide_states.safetensor")  # noqa: PTH118
+        states_file_path = os.path.join(directory_path, "tide_states.safetensor")
         states = safe_load(states_file_path)
-        with open(  # noqa: PTH123
-            os.path.join(directory_path, "tide_parameters.json")  # noqa: PTH118
+        with open(
+            os.path.join(directory_path, "tide_parameters.json")
         ) as parameters_file:
             parameters = json.load(parameters_file)

Also applies to: 280-280, 284-284, 296-297, 305-305, 307-308

applications/equitypricemodel/src/equitypricemodel/tide_data.py (1)

271-276: Remove unused noqa directives flagged by Ruff.

Ruff reports unused noqa pragmas at lines 271, 273, and 276 because the corresponding rules (E501, C901) are not enabled. Removing these will clean up the code.

🔎 Proposed cleanup

-            "decoder_continuous_features": 0,  # not using decoder_continuous_features for now  # noqa: E501
+            "decoder_continuous_features": 0,  # not using decoder_continuous_features for now
             "static_categorical_features": len(self.static_categorical_columns),
-            "static_continuous_features": 0,  # not using static_continuous_features for now  # noqa: E501
+            "static_continuous_features": 0,  # not using static_continuous_features for now
         }

-    def get_batches(  # noqa: C901
+    def get_batches(
         self,

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to **/*.py : Lint Python code with Ruff using the comprehensive ruleset

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to **/*.py : Use Ruff as the project formatter (replaces black/isort) for all Python code

Learnt from: forstmeier
Repo: pocketsizefund/pocketsizefund PR: 635
File: applications/datamanager/src/predictions.rs:60-99
Timestamp: 2025-09-24T00:54:42.461Z
Learning: In applications/datamanager/src/predictions.rs, there is a known SQL injection vulnerability where user-supplied tickers from predictions_query flow to storage::query_predictions_dataframe_from_s3 without validation, and the storage function builds SQL queries via string concatenation. The user forstmeier acknowledged this issue but chose to skip fixing it in PR #635, deferring it for later.

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to **/*.py : Use Structlog for structured logging across services

Learnt from: forstmeier
Repo: pocketsizefund/pocketsizefund PR: 624
File: applications/datamanager/src/routes/equity.rs:6-6
Timestamp: 2025-09-16T20:20:16.305Z
Learning: User forstmeier prefers to defer switching from `Option<Json<DateRangeQuery>>` to `Option<Query<DateRangeQuery>>` for GET endpoints in the pocketsizefund datamanager Rust application, even when it's technically more correct for RESTful APIs.

Learnt from: CR
Repo: pocketsizefund/pocketsizefund PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-09-16T16:54:51.913Z
Learning: Applies to {pyproject.toml,applications/*/pyproject.toml,libraries/python/pyproject.toml} : Enforce Python version 3.12.10 across all projects

maskfile.md:1

• The cron schedule '0 23 * * *' runs at 11:00 PM UTC, which is 6:00 PM EST during standard time but 7:00 PM EST during daylight saving time. Consider clarifying the timezone handling or adjusting for DST if the 6:00 PM EST time is critical.

# Pocket Size Fund Task Manager

tools/sync_equity_bars_data.py (2)

117-117: Remove unused noqa directive.

The # noqa: PLR2004 directive on line 117 is flagged by Ruff (RUF100) because the PLR2004 rule is not enabled. Remove the directive—the literal 3 is clear and acceptable for argument count validation.

🔎 Proposed fix

-    if len(sys.argv) != 3:  # noqa: PLR2004
+    if len(sys.argv) != 3:

Based on static analysis hints.

---

102-102: Remove unused noqa directive.

The # noqa: PLR2004 directive on line 102 is flagged by Ruff (RUF100) because the PLR2004 rule is not enabled in your configuration. Remove the directive—the magic number 400 is clear and acceptable in this context.

🔎 Proposed fix

-            if status_code >= 400:  # noqa: PLR2004
+            if status_code >= 400:

Based on static analysis hints.

applications/datamanager/src/storage.rs (2)

184-203: Ticker validation improves security but error message leaks input.

The ticker validation (lines 186-194) correctly addresses the SQL injection concerns from previous reviews. However, the error message at line 192 exposes the invalid ticker value, which could aid security probing.

🔎 Proposed fix (from past review)

-                    return Err(Error::Other(format!("Invalid ticker format: {}", ticker)));
+                    return Err(Error::Other("Invalid ticker format".to_string()));

This matches the past comment from Copilot about avoiding leaking the invalid input value.

---

323-323: Inconsistent debug log format.

Line 323 uses "Executing export SQL:" while lines 224 and 418 use "Executing query SQL:". Maintaining consistent log message formats improves grep-ability and observability.

🔎 Proposed fix

-    debug!("Executing export SQL: {}", query);
+    debug!("Executing query SQL: {}", query);

This matches the format used elsewhere in the file and aligns with the past Copilot review comment.

tools/sync_equity_bars_data.py (2)

12-56: Consider adding descriptive messages to RuntimeError exceptions.

The function raises RuntimeError without custom messages on lines 17, 21, 36, and 54. While the logger captures context, adding explicit exception messages would improve debugging and make the call stack more informative for callers who catch these exceptions.

🔎 Suggested improvements

     try:
         date_range = json.loads(date_range_json)
     except json.JSONDecodeError as e:
         logger.exception("JSON decoding error", error=f"{e}")
-        raise RuntimeError from e
+        raise RuntimeError("Failed to decode date range JSON") from e

     if "start_date" not in date_range or "end_date" not in date_range:
         logger.error("Missing required date fields", date_range=date_range)
-        raise RuntimeError
+        raise RuntimeError("Missing required date fields: start_date and end_date")

     try:
         start_date = datetime.strptime(date_range["start_date"], "%Y-%m-%d").replace(
             tzinfo=UTC
         )
         end_date = datetime.strptime(date_range["end_date"], "%Y-%m-%d").replace(
             tzinfo=UTC
         )
     except ValueError as e:
         logger.exception(
             "Date parsing error",
             error=f"{e}",
             required_format="YYYY-MM-DD",
         )
-        raise RuntimeError from e
+        raise RuntimeError("Failed to parse dates in YYYY-MM-DD format") from e

     # ... date clamping logic ...

     if start_date > end_date:
         logger.error(
             "Invalid date range after clamping",
             start_date=start_date.strftime("%Y-%m-%d"),
             end_date=end_date.strftime("%Y-%m-%d"),
         )
-        raise RuntimeError
+        raise RuntimeError(
+            "Invalid date range: start_date is after end_date following two-year window clamping"
+        )

---

127-138: Consider simplifying empty string validation.

The current implementation creates a dictionary solely for logging, then loops through both arguments. While the validation itself is valuable (preventing empty strings like python script.py "" "{}"), the logic could be more concise.

🔎 Alternative implementation

     base_url = sys.argv[1]
     raw_date_range = sys.argv[2]

-    arguments = {
-        "base_url": base_url,
-        "raw_date_range": raw_date_range,
-    }
-
-    for argument in [base_url, raw_date_range]:
-        if not argument:
-            logger.error(
-                "Missing required positional argument(s)",
-                **arguments,
-            )
-            sys.exit(1)
+    if not base_url or not raw_date_range:
+        logger.error(
+            "Empty argument(s) provided",
+            base_url=base_url or "(empty)",
+            raw_date_range=raw_date_range or "(empty)",
+        )
+        sys.exit(1)

Alternatively, keep the current approach if you prefer the explicit dictionary—it's clear and works correctly.

applications/equitypricemodel/src/equitypricemodel/server.py (5)

31-38: Add error handling and logging for model loading.

The endpoint lacks error handling and observability. If Model.load() fails (missing files, corrupt state), the endpoint will return a generic 500 error. The hard-coded directory path "." also reduces testability.

🔎 Proposed improvements

+logger = structlog.get_logger()
+
+MODEL_DIRECTORY = os.getenv("PSF_MODEL_DIRECTORY", ".")
+
 @application.post("/predictions")
 def create_predictions() -> PredictionResponse:
+    logger.info("predictions.create.started")
+    
+    try:
-        tide_model = Model.load(directory_path=".")
+        tide_model = Model.load(directory_path=MODEL_DIRECTORY)
+    except Exception as e:
+        logger.error("predictions.model_load.failed", error=str(e))
+        raise

Based on coding guidelines requiring Structlog for structured logging.

---

40-66: Past issues resolved; consider adding request logging.

The previous issues have been correctly addressed:

• raise_for_status() is now called after each request (lines 49, 56)
• pl.DataFrame() is correctly used instead of pl.read_json() (line 64)
• Data.load() classmethod usage is correct (line 89, reviewed separately)

However, consider adding structured logging for the HTTP requests to improve observability of external dependencies.

🔎 Optional logging additions

+    logger.info("predictions.fetch_equity_bars.started", 
+                start_date=start_date.isoformat(), 
+                end_date=end_date.isoformat())
     equity_bars_response = requests.get(
         url=f"{DATAMANAGER_BASE_URL}/equity-bars",
         params={
             "start_date": start_date.isoformat(),
             "end_date": end_date.isoformat(),
         },
         timeout=60,
     )
     equity_bars_response.raise_for_status()
+    logger.info("predictions.fetch_equity_bars.completed", 
+                bytes=len(equity_bars_response.content))

---

85-91: Validate non-empty data before processing.

After joining and selecting columns (line 85), the data DataFrame could be empty if no tickers match between equity_details and equity_bars. Processing empty data through the TiDE pipeline will likely fail in unexpected ways.

🔎 Proposed validation

     data = consolidated_data.select(retained_columns)
+    
+    if data.height == 0:
+        logger.error("predictions.data_consolidation.empty_result")
+        raise ValueError("No matching equity data found for prediction")
     
     current_timestamp = datetime.now(tz=UTC)

---

105-116: Validate filtered predictions are non-empty.

After filtering to the 7th timestep (line 107), processed_predictions could be empty if the model output doesn't include that specific day. Attempting to save and return empty predictions may not be the intended behavior.

🔎 Proposed validation

     processed_predictions = predictions.filter(
         pl.col("timestamp")
         == int(
             processed_prediction_timestamp.replace(
                 hour=0, minute=0, second=0, microsecond=0
             ).timestamp()
         )
     )
+    
+    if processed_predictions.height == 0:
+        logger.warning("predictions.filter.no_results", 
+                      target_timestamp=processed_prediction_timestamp.isoformat())
+        # Consider: should this be an error or return empty gracefully?
     
     processed_predictions = predictions_schema.validate(processed_predictions)

---

118-129: Add logging for prediction save operation.

The save operation to the datamanager lacks logging, making it difficult to observe whether predictions were successfully persisted.

🔎 Proposed logging

+    logger.info("predictions.save.started", 
+                count=processed_predictions.height,
+                timestamp=current_timestamp.isoformat())
     save_predictions_response = requests.post(
         url=f"{DATAMANAGER_BASE_URL}/predictions",
         json={
             "timestamp": current_timestamp.isoformat(),
             "data": processed_predictions.to_dicts(),
         },
         timeout=60,
     )
     
     save_predictions_response.raise_for_status()
+    logger.info("predictions.save.completed")
     
     return PredictionResponse(data=processed_predictions.to_dict())

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: forstmeier
Repo: pocketsizefund/pocketsizefund PR: 635
File: applications/datamanager/src/predictions.rs:60-99
Timestamp: 2025-09-24T00:54:42.461Z
Learning: In applications/datamanager/src/predictions.rs, there is a known SQL injection vulnerability where user-supplied tickers from predictions_query flow to storage::query_predictions_dataframe_from_s3 without validation, and the storage function builds SQL queries via string concatenation. The user forstmeier acknowledged this issue but chose to skip fixing it in PR #635, deferring it for later.

applications/equitypricemodel/src/equitypricemodel/server.py (1)

49-57: Narrow exception handling to specific exception types.

Catching generic Exception is too broad and may mask unexpected errors. Use requests.exceptions.RequestException (or more specific subclasses like requests.exceptions.HTTPError, requests.exceptions.Timeout) to handle only request-related failures.

Additionally, note that line 140 raises the exception while other error paths return HTTP 500 responses — consider returning a 500 status consistently instead of propagating the exception.

Details

🔎 Suggested improvements

+import requests.exceptions

-    except Exception as e:
+    except requests.exceptions.RequestException as e:
         logger.exception(
             "Failed to fetch equity bars data",
             start_date=start_date.isoformat(),
             end_date=end_date.isoformat(),
             error=f"{e}",
         )
 
         return Response(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)

Apply similar changes to the other two try-except blocks (lines 67-73 and 134-140). For consistency at line 140, return an error response instead of raising:

-    except Exception as e:
+    except requests.exceptions.RequestException as e:
         logger.exception(
             "Failed to save predictions data",
             timestamp=current_timestamp.isoformat(),
             error=f"{e}",
         )
-        raise
+        return Response(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)

Also applies to: 67-73, 134-140

infrastructure/__main__.py (1)

564-663: Consider adding task roles to portfoliomanager and equitypricemodel for consistency.

While these services don't currently require AWS resource access (portfoliomanager only needs Alpaca API access via secrets, equitypricemodel only calls datamanager), adding task roles now (even with minimal/no permissions) would provide:

• Consistency with datamanager's configuration
• Easier future extensibility if AWS resource access is needed
• Clearer separation between execution permissions and runtime permissions

🔎 Example: Add task role to portfoliomanager

 portfoliomanager_task_definition = aws.ecs.TaskDefinition(
     "portfoliomanager_task",
     family="portfoliomanager",
     cpu="256",
     memory="512",
     network_mode="awsvpc",
     requires_compatibilities=["FARGATE"],
     execution_role_arn=execution_role.arn,
+    task_role_arn=task_role.arn,
     container_definitions=pulumi.Output.all(

Apply the same change to equitypricemodel_task_definition at line 631.

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

applications/equitypricemodel/src/equitypricemodel/server.py (2)

38-86: Consider more specific exception handling for better error diagnostics.

The try-except blocks currently catch generic Exception, which can mask unexpected errors. Consider catching specific exception types:

• Lines 38-58 and 60-74: Catch requests.exceptions.RequestException for HTTP-related failures
• Lines 76-86: Catch specific exceptions from parse_responses (e.g., pl.exceptions.PolarsError, ValueError)

This allows unexpected exceptions to propagate naturally while handling known failure modes explicitly.

🔎 Example refactor

     try:
         equity_bars_response = requests.get(
             url=f"{DATAMANAGER_BASE_URL}/equity-bars",
             params={
                 "start_date": start_date.isoformat(),
                 "end_date": end_date.isoformat(),
             },
             timeout=60,
         )
 
         equity_bars_response.raise_for_status()
 
-    except Exception as e:
+    except requests.exceptions.RequestException as e:
         logger.exception(
             "Failed to fetch equity bars data",
             start_date=start_date.isoformat(),
             end_date=end_date.isoformat(),
             error=f"{e}",
         )
 
         return Response(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)

Apply similar changes to the other exception handlers.

---

145-148: Consider using JSONResponse for more idiomatic FastAPI code.

While the current implementation correctly encodes the JSON response as bytes, using JSONResponse would be cleaner and more consistent with FastAPI conventions.

🔎 Proposed refactor

Update the import:

-from fastapi import FastAPI, Response, status
+from fastapi import FastAPI, Response, status
+from fastapi.responses import JSONResponse

Simplify the return statement:

-    return Response(
-        content=json.dumps({"data": processed_predictions.to_dict()}).encode("utf-8"),
-        status_code=status.HTTP_200_OK,
-    )
+    return JSONResponse(
+        content={"data": processed_predictions.to_dict()},
+        status_code=status.HTTP_200_OK,
+    )

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro